 #define _GNU_SOURCE             /* See feature_test_macros(7) */
#include <sys/types.h>
#include <unistd.h>
#include <pthread.h>
#include <sys/un.h>
#include <sys/socket.h>
#include <errno.h>
#include <sys/epoll.h>
#include <stdio.h>
#include <sys/time.h>
#include <stdlib.h>
#include <stdatomic.h>


#include "somb_daemon.pb.h"

#include "somb.h"
#include "somb_manager.h"
#include "somb_log.h"
#include "somb_message.h"

#include "somb_work.h"
#include "pthread.h"

#include "somb_lib.h"
#include "somb_sock.h"
#include "message_queue.h"

#define SOMB_WORK_POLLWAIT_ITEM_MAX  10
struct somb_worker_st
{
    pthread_t tid;
    uint8_t wid;
    uint32_t msg_count;
    uint64_t time;
    struct message_queue msgque;
};

static struct somb_worker_st gstSombWork[SOMB_WORKER_MAX_THREAD];

static uint32_t balance_worker_id=0;

void *vsomb_worker_thread(void *arg)
{
    //int ret = SOMB_OK;
    struct somb_worker_st *pstsombworker = (struct somb_worker_st *)arg;
    uint8_t workid = pstsombworker->wid;

    struct somb_local_sub_ns *pstLocalSubNs = pstSOMB_GetLocalSubNs();
    if(NULL == pstLocalSubNs)
    {
        SOMB_LOG_ERR("somb work:pid(%d) worker(%d) get local sub ns failled",getpid(),pstsombworker->wid);
        return NULL;
    }

    while(1)
    {
        struct somb_worker_msg *workmsg = message_queue_read(&(gstSombWork[workid].msgque));
        if(NULL == workmsg)
        {
            SOMB_LOG_ERR("somb work:(%s) worker(%d) recv null msg",pstLocalSubNs->localexename,workid);
            continue;
        }

        //uint32_t src_sub_ns_id =0 ;
        //sscanf(workmsg->addr,SOMB_SUB_NS_ADDR_FMT1,&src_sub_ns_id);


        uint32_t handle = ((workmsg->MsgType << 24)&0xFF000000)\
                          | ((workmsg->SeqId << 16)&0x00FF0000)\
                          | workmsg->sub_ns_id;

        SOMB_LOG_DEBUG("somb work:(%s) worker(%d) recv Seqid(%d) ServiceId(%08x) Msgtype(%d) from subns(%d) handle=%08x",pstLocalSubNs->localexename,workid,workmsg->SeqId,workmsg->ServiceId,workmsg->MsgType,workmsg->sub_ns_id,handle);
        struct timeval start;
        gettimeofday(&start,NULL);
        uint32_t *time_local = (uint32_t *)&(gstSombWork[workid].time);
        time_local[0]=start.tv_sec;
        time_local[1]=start.tv_usec;

        switch(workmsg->MsgType)
        {
            case SOMB_MSG_TYPE_METHOD_RESPONSE_SYNC:
                    //response sync should be handled by work proxy thread!
                    SOMB_LOG_ERR("somb work:(%s) worker(%d) ServiceId(%08x) msg type(%d) method response sync should not be handled here",pstLocalSubNs->localexename,workid,workmsg->ServiceId,workmsg->MsgType);
                    break;
            case SOMB_MSG_TYPE_METHOD_RESPONSE_ASYNC:
                {
                    PF_SOMB_METHOD_ASYNC_CB pFunc = pfSOMB_Manager_Get_Method_AsyncCB(workmsg->ServiceId);
                    if(NULL == pFunc)
                    {
                        SOMB_LOG_ERR("somb work:(%s) worker(%d) ServiceId(%08x) msg type(%d) method async cb =NULL",pstLocalSubNs->localexename,workid,workmsg->ServiceId,workmsg->MsgType);
                    }
                    else
                    {
                        pFunc(workmsg->ServiceId,workmsg->data,workmsg->ServicePayloadLen);
                    }
                }
                break;
            case SOMB_MSG_TYPE_METHOD_REQUEST_SYNC:
            case SOMB_MSG_TYPE_METHOD_REQUEST_ASYNC:
            case SOMB_MSG_TYPE_METHOD_SETREQUEST:
                {
                    ST_SOMB_LIB_SERVICE_LIST *pstServiceInfo =  pstSOMBManagerPureGetServiceInfo(workmsg->ServiceId);
                    if(NULL == pstServiceInfo)
                    {
                        SOMB_LOG_ERR("somb work:(%s) worker(%d) get ServiceId(%08x) from manager failed",pstLocalSubNs->localexename,workid,workmsg->ServiceId);
                        break;
                    }

                    atomic_fetch_add(&(pstServiceInfo->ServiceCount),1);
                    PF_SOMB_METHOD_PROVIDER pFunc = (PF_SOMB_METHOD_PROVIDER)pstServiceInfo->ServicePriv;
                    if(NULL != pFunc)
                    {
                        //pFunc(src_sub_ns_id,ServiceId,payload,workmsg->ServicePayloadLen);
                        pFunc(handle,workmsg->ServiceId,workmsg->data,workmsg->ServicePayloadLen);
                    }
                    else
                    {
                        SOMB_LOG_ERR("somb work:(%s) worker(%d) ServiceId(%08x) msg type(%d) method provider =NULL",pstLocalSubNs->localexename,workid,workmsg->ServiceId,workmsg->MsgType);
                    }
                }
                break;
            case SOMB_MSG_TYPE_EVENT:
                {
                    ST_SOMB_LIB_SERVICE_LIST *pstServiceInfo =  pstSOMBManagerPureGetServiceInfo(workmsg->ServiceId);
                    if(NULL == pstServiceInfo)
                    {
                        SOMB_LOG_ERR("somb work:(%s) worker(%d) get ServiceId(%08x) from manager failed",pstLocalSubNs->localexename,workid,workmsg->ServiceId);
                        break;
                    }
                    atomic_fetch_add(&(pstServiceInfo->ServiceCount),1);
                    PF_SOMB_EVENT_LISTENER_CB pFunc = (PF_SOMB_EVENT_LISTENER_CB)pstServiceInfo->ServicePriv;
                    if(NULL != pFunc)
                    {
                        pFunc(workmsg->ServiceId,workmsg->data,workmsg->ServicePayloadLen);
                    }
                    else
                    {
                        SOMB_LOG_ERR("somb work:(%s) worker(%d) ServiceId(%08x) msg type(%d) event listen cb =NULL",pstLocalSubNs->localexename,workid,workmsg->ServiceId,workmsg->MsgType);
                    }
                }
                break;
            default:
                SOMB_LOG_ERR("somb work:(%s) worker(%d) ServiceId(%08x) msg type(%d) unsupported ",pstLocalSubNs->localexename,workid,workmsg->ServiceId,workmsg->MsgType);
                break;
        }

        if(workmsg->data != NULL)
            free(workmsg->data);

        message_queue_message_free(&(gstSombWork[workid].msgque),workmsg);

        gstSombWork[workid].msg_count++;
    }
}

void vsomb_worker_proxy_sk1_process(int epoll_fd,struct somb_local_sub_ns *pstLocalSubNs)
{
    struct sockaddr_un addr_un;
    memset(&addr_un,0,sizeof(struct sockaddr_un));
    socklen_t socklen =sizeof(struct sockaddr_un);

    int c_fd = accept4(pstLocalSubNs->sk1, (struct sockaddr*)&addr_un, (socklen_t *)&socklen,SOCK_CLOEXEC);
    if(c_fd < 0)
    {
       SOMB_LOG_ERR("somb work:proxy (%s) subns id(%d) accetp failed,sk1=%d",pstLocalSubNs->localexename,pstLocalSubNs->sub_ns_id,pstLocalSubNs->sk1);
       return ;
    }

    if(NULL != strstr(addr_un.sun_path,"somb_ctrl"))
    {
        SOMB_LOG_DEBUG("somb work:proxy (%s) subns id(%d) daemon-ctrl connected",pstLocalSubNs->localexename,pstLocalSubNs->sub_ns_id);
        pstLocalSubNs->ctrl_sk = c_fd;

        struct epoll_event e_event;
        e_event.data.fd = c_fd;
        e_event.events = EPOLLIN | EPOLLRDHUP;
        int ret = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, c_fd, &e_event);
        if(ret < 0)
        {
           SOMB_LOG_ERR("somb work:proxy (%s) subns id(%d) add ctrl fd=%d to epoll failed",pstLocalSubNs->localexename,pstLocalSubNs->sub_ns_id,c_fd);
        }
        return;
    }

    SOMB_LOG_DEBUG("somb work:proxy (%s) subns id(%d) accetp new client=%d from(%s)",pstLocalSubNs->localexename,pstLocalSubNs->sub_ns_id,c_fd,addr_un.sun_path);
    uint32_t sub_ns_id = 0,my_sub_ns_id=0;
    sscanf(addr_un.sun_path,SOMB_SUB_NS_ADDR_FMT2,&sub_ns_id,&my_sub_ns_id);
    uint8_t src_addr[SOMB_ADDR_LEN]={0};
    sprintf(src_addr,SOMB_SUB_NS_ADDR_FMT1,sub_ns_id);
    SOMB_LOG_DEBUG("somb work:proxy (%s) subns id(%d)  sub_ns_id(%d) my_sub_ns_id(%d)",\
                   pstLocalSubNs->localexename,pstLocalSubNs->sub_ns_id,\
                   sub_ns_id, my_sub_ns_id);


    struct somb_mngr_sub_ns *pstSubNs = pstSOMB_Manager_Find_SubNs(sub_ns_id);
    if(NULL == pstSubNs)
    {
        pstSubNs = pstSOMB_Manager_SaveSubNs(sub_ns_id,src_addr,NULL);
        if(NULL == pstSubNs)
        {
            SOMB_LOG_ERR("somb work:proxy (%s) subns id(%d) accetp new client=%d from(%s) but not exist in SubNsList,add failed",pstLocalSubNs->localexename,pstLocalSubNs->sub_ns_id,c_fd,addr_un.sun_path);
            return ;
        }
        else
        {
            SOMB_LOG_WARN("somb work:proxy (%s) subns id(%d) accetp new client=%d from(%s) but not exist in SubNsList,add success",pstLocalSubNs->localexename,pstLocalSubNs->sub_ns_id,c_fd,addr_un.sun_path);
            pstSubNs->skfd2 = c_fd;
        }
    }


    struct epoll_event e_event;
    e_event.data.fd = c_fd;
    //e_event.data.ptr= (void *)pstSubNs;
    //e_event.events = EPOLLIN | EPOLLRDHUP | EPOLLPRI;
    e_event.events = EPOLLIN | EPOLLRDHUP;
    int ret = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, c_fd, &e_event);
    if(ret < 0)
    {
       SOMB_LOG_ERR("somb work:proxy (%s) subns id(%d) add new client=%d to epoll failed",pstLocalSubNs->localexename,pstLocalSubNs->sub_ns_id,c_fd);
    }
}
void vsomb_worker_proxy_business_process(int epoll_fd,struct somb_local_sub_ns *pstLocalSubNs,int clientfd )
{
    int ret;
    uint32_t ServiceId;
    uint32_t ServicePayloadLen;
    uint8_t MsgType;
    uint8_t SeqId;
    uint8_t headbuffer[SOMB_MESSAGE_HEADER_SIZE];
    int recvlen = -1;
    //struct msghdr msg;
    //struct iovec iov[1];
    memset(headbuffer,0,sizeof(headbuffer));

    recvlen = recv(clientfd,headbuffer,SOMB_MESSAGE_HEADER_SIZE,0);
    if(0 == recvlen)
    {
        SOMB_LOG_WARN("somb work-proxy:(%s) SubNsId(%d) pid(%d) del client fd(%d)  ",pstLocalSubNs->localexename, pstLocalSubNs->sub_ns_id,pstLocalSubNs->localpid, clientfd);
        ret = epoll_ctl(epoll_fd, EPOLL_CTL_DEL, clientfd, NULL);
        if(ret != 0)
        {
            SOMB_LOG_ERR("somb work-proxy:(%s) SubNsId(%d) pid(%d) del new client fd(%d) to epoll failed",pstLocalSubNs->localexename, pstLocalSubNs->sub_ns_id,pstLocalSubNs->localpid, clientfd);
        }

        struct somb_mngr_sub_ns *pstSubNs = pstSOMB_GetSubNsBySkfd2(clientfd);
        if(NULL != pstSubNs)
        {
            pstSubNs->SubNsStatus = SOMB_SUBNS_STATUS_OFFLINE;
        }
        else
        {
            SOMB_LOG_ERR("somb work-proxy:(%s) SubNsId(%d) pid(%d) find client fd(%d) failed",pstLocalSubNs->localexename, pstLocalSubNs->sub_ns_id,pstLocalSubNs->localpid, clientfd);
        }
        return ;
    }
#if 0
    struct sockaddr_un addr_un;
    memset(&addr_un,0,sizeof(struct sockaddr_un));
    socklen_t socklen =sizeof(struct sockaddr_un);
    recvlen = recvfrom(pstLocalSubNs->sk1,headbuffer,sizeof(headbuffer),MSG_DONTWAIT,(struct sockaddr *)&addr_un,&socklen);
    if(recvlen != sizeof(headbuffer))
    {
       SOMB_LOG_ERR("somb work:proxy (%s) subns id(%d) recvlen=%d",pstLocalSubNs->localexename,pstLocalSubNs->sub_ns_id,recvlen);
       return;
    }
//#else
    bzero(&msg,sizeof(struct msghdr));
    bzero(iov,sizeof(struct iovec));
    iov[0].iov_base =  headbuffer;
    iov[0].iov_len=  SOMB_MESSAGE_HEADER_SIZE;
    msg.msg_name = &addr_un;
    msg.msg_namelen = sizeof(struct sockaddr_un);
    msg.msg_iov = iov;
    msg.msg_iovlen = 1;
    ret = recvmsg(pstLocalSubNs->sk1, &msg, 0);
    if(ret < 0)
    {
       SOMB_LOG_ERR("somb work:proxy (%s) subns id(%d) recvlen=%ld",pstLocalSubNs->localexename,pstLocalSubNs->sub_ns_id,msg.msg_iov[0].iov_len);
        return;
    }
    recvlen= msg.msg_iov[0].iov_len;
    SOMB_LOG_DEBUG("somb work:proxy (%s) subns id(%d) recv msg from %s ",pstLocalSubNs->localexename,pstLocalSubNs->sub_ns_id,addr_un.sun_path);
#endif



    SeqId = *(uint8_t*)&(headbuffer[2]);
    MsgType = *(uint8_t*)&(headbuffer[3]);
    ServiceId = (headbuffer[4] << 24)|(headbuffer[5] << 16)|(headbuffer[6] << 8)|(headbuffer[7]);
    //ServicePayloadLen= *(uint32_t*)&(headbuffer[8]);
    ServicePayloadLen = (headbuffer[8] << 24)|(headbuffer[9] << 16)|(headbuffer[10] << 8)|(headbuffer[11]);


    uint32_t src_sub_ns_id = (headbuffer[12] << 24)|(headbuffer[13] << 16)|(headbuffer[14] << 8)|(headbuffer[15]);
    SOMB_LOG_DEBUG("somb work:proxy (%s) subns id(%d) recv msg from clientfd(%d) subns(%d)",pstLocalSubNs->localexename,pstLocalSubNs->sub_ns_id,clientfd,src_sub_ns_id);

    SOMB_LOG_DEBUG("somb work:proxy (%s) subns id(%d) ServiceId(%08x) MsgType(%d) SeqId(%d) PLen(%d)",\
                   pstLocalSubNs->localexename,pstLocalSubNs->sub_ns_id,ServiceId,MsgType,SeqId,ServicePayloadLen);

    if(SOMB_MSG_TYPE_METHOD_RESPONSE_SYNC == MsgType)
    {
        void *buffer = NULL;
        if((ServicePayloadLen > 0)&&(0xFDFEFAFD != ServicePayloadLen))
        {
            buffer = malloc(ServicePayloadLen);
            if(NULL == buffer)
            {
                SOMB_LOG_ERR("somb work:proxy (%s) subns id(%d) malloc for buffer failed,ServiceId=%08x MsgSeqId=%d",pstLocalSubNs->localexename,pstLocalSubNs->sub_ns_id,ServiceId,SeqId);
                return ;
            }

            bzero(buffer,ServicePayloadLen);

            recvlen = recvfrom(clientfd,buffer,ServicePayloadLen,MSG_DONTWAIT,NULL,NULL);
            if(-1 == recvlen)
            {
                SOMB_LOG_ERR("somb work:proxy (%s) subns id(%d) recv len=%d err=%s",pstLocalSubNs->localexename,pstLocalSubNs->sub_ns_id,recvlen,strerror(errno));
                free(buffer);
                return;
            }
        }
        i32SOMB_Msg_Resp_Sync_Complete(ServiceId,SeqId,&buffer,ServicePayloadLen);
        return;
    }

    //SOMB_MESSAGE_HEADER_DUMP(headbuffer); 
    uint8_t workid = balance_worker_id%(SOMB_WORKER_MAX_THREAD-1) + 1;

    struct somb_worker_msg *workmsg=NULL;
    struct message_queue *pMsgQue= NULL;
    pMsgQue = &(gstSombWork[workid].msgque);
    workmsg = message_queue_message_alloc_blocking(pMsgQue);
    if(NULL == workmsg)
    {
        SOMB_LOG_ERR("somb work:proxy (%s) subns id(%d) malloc msg queue data for worker(%d) failed",pstLocalSubNs->localexename,pstLocalSubNs->sub_ns_id,workid);
        return ;
    }
   
    int32_t recvlen1=0;

    workmsg->ServiceId = ServiceId;
    workmsg->MsgType = MsgType;
    workmsg->SeqId = SeqId;
    workmsg->sub_ns_id = src_sub_ns_id;
    //if((0 != ServicePayloadLen)&&(0xFDFEFAFD != ServicePayloadLen))
    if(0 < ServicePayloadLen)
    {
        workmsg->ServicePayloadLen = ServicePayloadLen;
        workmsg->data = malloc(ServicePayloadLen);

        if(workmsg->data == NULL)
        {
            SOMB_LOG_ERR("somb work:proxy (%s) subns id(%d) malloc %d for payload failed",pstLocalSubNs->localexename,pstLocalSubNs->sub_ns_id,ServicePayloadLen);
            message_queue_message_free(pMsgQue,workmsg);
            return ;
        }

        memset(workmsg->data,0,ServicePayloadLen);
        recvlen1 = recv(clientfd,workmsg->data,ServicePayloadLen,0);
        if((-1 == recvlen1)||(ServicePayloadLen !=  recvlen1))
        {
            SOMB_LOG_ERR("somb work:proxy (%s) subns id(%d) recv len=%d recvlen1=%d err=%s",pstLocalSubNs->localexename,pstLocalSubNs->sub_ns_id,ServicePayloadLen,recvlen1,strerror(errno));
            free(workmsg->data);
            message_queue_message_free(pMsgQue,workmsg);
            return ;
        }
    }
    else if(0  == ServicePayloadLen)
    {
        workmsg->ServicePayloadLen = 0;
        //workmsg->data = malloc(1024); //this should not be accessed
        //memset(workmsg->data,0,1024);
        workmsg->data = NULL;
    }
    else {
        SOMB_LOG_ERR("somb work:proxy (%s) subns id(%d) ServiceId(%08x) MsgType(%d) SeqId(%d) PLen(%d)",pstLocalSubNs->localexename,pstLocalSubNs->sub_ns_id,ServiceId,MsgType,SeqId,ServicePayloadLen);
    }
    //sprintf(workmsg->addr,"%s",addr_un.sun_path);

    SOMB_LOG_DEBUG("somb work:proxy (%s) subns id(%d) recv SeqId(%d) ServiceId(%08x) MsgType(%d) PayloadLen=%d recvlen=%d recvlen1=%d push to worker(%d) from subnsid(%d)",pstLocalSubNs->localexename,pstLocalSubNs->sub_ns_id,SeqId,ServiceId,MsgType,ServicePayloadLen,recvlen,recvlen1,workid,src_sub_ns_id);

    message_queue_write(pMsgQue,workmsg);
    balance_worker_id++;

}

void vsomb_worker_get_subns_status_to_ctrl(int fd)
{
    somb_subns_work_status work_status = somb_subns_work_status_init_zero;

    vsomb_manager_get_service_statistic(&work_status);

    work_status.work_job_last_time_count = 3;
    work_status.work_job_num_count = 3;

    int loop = 0;
    for(loop=0;loop<3;loop++)
    {
        work_status.work_job_last_time[loop] = gstSombWork[loop+1].time;
        work_status.work_job_num[loop] = gstSombWork[loop+1].msg_count;
    }

    uint8_t buffer[SOMB_DAEMON_MSG_HEADER_SIZE+somb_subns_work_status_size];
    memset(buffer,0,sizeof(buffer));



    pb_ostream_t stream = pb_ostream_from_buffer(&(buffer[SOMB_DAEMON_MSG_HEADER_SIZE]), somb_subns_work_status_size);
    bool status = pb_encode(&stream,somb_subns_work_status_fields ,&work_status);
    if(!status)
    {
        SOMB_LOG_ERR("somb sock:pid(%d) send to ctrl encode failed,errno=%s",getpid(),PB_GET_ERROR(&stream));
    }

    *(uint32_t*)buffer = somb_daemon_msgid_ctrl_get_sub_ns_status;
    *(uint32_t*)(buffer+4) = stream.bytes_written;

    int ret  = send(fd,buffer,SOMB_DAEMON_MSG_HEADER_SIZE+stream.bytes_written,0);
    if(-1 == ret)
    {
        SOMB_LOG_ERR("somb sock:pid(%d) send to ctrl failed,errno=%s",getpid(),strerror(errno));
    }

}

void vsomb_worker_proxy_ctrl_process(int epoll_fd,struct somb_local_sub_ns *pstLocalSubNs)
{
    int recvlen =0;
    uint8_t buffer[SOMB_MSG_MAX_LENGTH];
    struct sockaddr_un addr_un;

    memset(buffer,0,sizeof(buffer));
    memset(&addr_un,0,sizeof(struct sockaddr_un));
    socklen_t socklen =sizeof(struct sockaddr_un);


    recvlen = recvfrom(pstLocalSubNs->ctrl_sk,buffer,sizeof(buffer),MSG_DONTWAIT,(struct sockaddr *)&addr_un,&socklen);
    if(recvlen == 0)
    {
        pstLocalSubNs->ctrl_sk = 0;
        epoll_ctl(epoll_fd, EPOLL_CTL_DEL, pstLocalSubNs->ctrl_sk, NULL);
        return ;
    }

    if(recvlen < SOMB_DAEMON_MSG_HEADER_SIZE)
    {
       SOMB_LOG_ERR("somb work:proxy-ctrl(%s) subns id(%d) recvlen=%d",pstLocalSubNs->localexename,pstLocalSubNs->sub_ns_id,recvlen);
       return ;
    }

    SOMB_LOG_DEBUG("somb work:proxy-ctrl(%s) subns(%d) recv TotalLen=%d from(%s)",pstLocalSubNs->localexename,pstLocalSubNs->sub_ns_id,recvlen,addr_un.sun_path);

}

void vsomb_worker_proxy_daemon_or_ctrl_process(int fd,int epoll_fd,struct somb_local_sub_ns *pstLocalSubNs)
{
     //we found issues: during start, the message from somb-daemon will be larger than SOMB_MSG_MAX_LENGTH(4096)
     //so we extend this buffer to 3 times of SOMB_MSG_MAX_LENGTH
    uint8_t buffer[SOMB_MSG_MAX_LENGTH*3]; 
    struct sockaddr_un addr_un;

    memset(buffer,0,sizeof(buffer));
    memset(&addr_un,0,sizeof(struct sockaddr_un));
    socklen_t socklen =sizeof(struct sockaddr_un);

    int recvlen = recvfrom(fd,buffer,sizeof(buffer),MSG_DONTWAIT,(struct sockaddr *)&addr_un,&socklen);
    if(recvlen == 0)
    {
        epoll_ctl(epoll_fd, EPOLL_CTL_DEL, fd, NULL);
        if(fd == pstLocalSubNs->sk)
        {
            SOMB_LOG_WARN("somb work:proxy-ctrl disconnecting with somb-daemon");
            pstLocalSubNs->sk = -1;
        }
        else if(fd == pstLocalSubNs->ctrl_sk)
        {
            SOMB_LOG_WARN("somb work:proxy-ctrl disconnecting with somb-ctrl");
            pstLocalSubNs->ctrl_sk = -1;
        }

        return ;
    }

    if(recvlen < SOMB_DAEMON_MSG_HEADER_SIZE)
    {
       SOMB_LOG_ERR("somb work:proxy-sk(%s) subns id(%d) recvlen=%d",pstLocalSubNs->localexename,pstLocalSubNs->sub_ns_id,recvlen);
       return ;
    }

    SOMB_LOG_DEBUG("somb work:proxy-sk(%s) subns(%d) recv TotalLen=%d from(%s)",pstLocalSubNs->localexename,pstLocalSubNs->sub_ns_id,recvlen,addr_un.sun_path);

    uint32_t MsgId;
    uint32_t MsgLen=0;
    uint32_t processed_msg_len=0;
    uint8_t *tmpbuffer = buffer;
    do
    {
        MsgId = *(uint32_t *)tmpbuffer;
        MsgLen = *(uint32_t *)(tmpbuffer + 4);

        SOMB_LOG_DEBUG("somb work:proxy-sk (%s) subns id(%d) recv len=%d msgid=%d processed_msg_len(%d)",pstLocalSubNs->localexename,pstLocalSubNs->sub_ns_id,MsgLen,MsgId,processed_msg_len);
        switch(MsgId)
        {
           case somb_daemon_msgid_get_sub_ns:
                {
                    vSOMB_Manager_Resp_Process(MsgId,somb_msg_get_sub_ns_resp_fields,somb_msg_get_sub_ns_resp_size,\
                            &(tmpbuffer[SOMB_DAEMON_MSG_HEADER_SIZE]),MsgLen);
                }
                break;
           case somb_daemon_msgid_register_sub_ns:
                {
                    vSOMB_Manager_Resp_Process(MsgId,somb_sub_ns_reg_resp_fields,somb_sub_ns_reg_resp_size,\
                            &(tmpbuffer[SOMB_DAEMON_MSG_HEADER_SIZE]),MsgLen);
                }
                break;
           case somb_daemon_msgid_get_service:
                {
                    vSOMB_Manager_Resp_Process(MsgId,somb_msg_get_service_resp_fields,somb_msg_get_service_resp_size,\
                            &(tmpbuffer[SOMB_DAEMON_MSG_HEADER_SIZE]),MsgLen);
                }
                break;
           case somb_daemon_msgid_listen_event_2_subns:
                {
                    vSOMB_Manager_Recv_Listen_SubNs_Process(&(tmpbuffer[SOMB_DAEMON_MSG_HEADER_SIZE]),MsgLen);
                }
                break;
            case somb_daemon_msgid_service_available_2_subns:
                vSOMB_Manager_Report_Service_OnAvailable(&(tmpbuffer[SOMB_DAEMON_MSG_HEADER_SIZE]),MsgLen);
                break;
            case somb_daemon_msgid_ctrl_get_sub_ns_status:
                vsomb_worker_get_subns_status_to_ctrl(fd);
                break;
            default:
                SOMB_LOG_ERR("somb work:proxy-sk (%s) subns id(%d) recv unknown msgid=%d",pstLocalSubNs->localexename,pstLocalSubNs->sub_ns_id,MsgId);
                break;
        }

        processed_msg_len += (SOMB_DAEMON_MSG_HEADER_SIZE + MsgLen);
        tmpbuffer = buffer + processed_msg_len;
    }while(processed_msg_len < recvlen);
}


void *vsomb_worker_proxy(void *arg)
{
    int ret = SOMB_OK;

    struct somb_local_sub_ns *pstLocalSubNs = pstSOMB_GetLocalSubNs();
    if(NULL == pstLocalSubNs)
    {
        SOMB_LOG_ERR("somb work-proxy:pid(%d) proxy get local sub ns failled",getpid());
        return NULL;
    }

    pthread_attr_t attr;
    pthread_attr_init(&attr);

    pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);

    uint32_t loop = 1;
    for(;loop<SOMB_WORKER_MAX_THREAD;loop++)
    {
        gstSombWork[loop].wid = loop;
        ret = message_queue_init(&(gstSombWork[loop].msgque), sizeof(struct somb_worker_msg), 20);
        if(0 != ret)
        {
            SOMB_LOG_ERR("somb work-proxy:pid(%d) create worker[%d]'s msg queue failled",getpid(),loop);
            ret = SOMB_FAIL;
            break;
        }

        ret = pthread_create(&(gstSombWork[loop].tid), &attr,vsomb_worker_thread,&(gstSombWork[loop]));
        if(0 != ret)
        {
            SOMB_LOG_ERR("somb work-proxy:pid(%d) create worker[%d] thread failled",getpid(),loop);
            ret = SOMB_FAIL;
            break;
        }
    }

    if(SOMB_OK != ret)
    {
        SOMB_LOG_ERR("somb work-proxy:(%s) SubNsId(%d) pid(%d) create worker failled",pstLocalSubNs->localexename,
                pstLocalSubNs->sub_ns_id,pstLocalSubNs->localpid);
        return NULL;
    }

    int epoll_fd = epoll_create(1); //size must be greater than 0
    if(epoll_fd < 0)
    {
        SOMB_LOG_ERR("somb work-proxy:(%s) SubNsId(%d) pid(%d) create epoll fd failled",pstLocalSubNs->localexename,
                pstLocalSubNs->sub_ns_id,pstLocalSubNs->localpid);
        return NULL;
    }

    struct epoll_event e_event;
    e_event.data.fd = pstLocalSubNs->sk1;     
    e_event.events = EPOLLIN;
	ret = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, pstLocalSubNs->sk1, &e_event);
	if(ret < 0)
	{
        SOMB_LOG_ERR("somb work-proxy:(%s) SubNsId(%d) pid(%d) add sk1 to epoll fd failled",pstLocalSubNs->localexename,
                pstLocalSubNs->sub_ns_id,pstLocalSubNs->localpid);
        return NULL;
	}
    e_event.data.fd = pstLocalSubNs->sk;     
    e_event.events = EPOLLIN;
	ret = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, pstLocalSubNs->sk, &e_event);
	if(ret < 0)
	{
        SOMB_LOG_ERR("somb work-proxy:(%s) SubNsId(%d) pid(%d) add sk to epoll fd failled",pstLocalSubNs->localexename,
                pstLocalSubNs->sub_ns_id,pstLocalSubNs->localpid);
        return NULL;
	}
    
    SOMB_LOG_INFO("somb work-proxy:(%s) SubNsId(%d) pid(%d) start Ok! ",pstLocalSubNs->localexename,
                pstLocalSubNs->sub_ns_id,pstLocalSubNs->localpid);

    while(1)
    {
        struct epoll_event wait_event[SOMB_WORK_POLLWAIT_ITEM_MAX];  
        ret = epoll_wait(epoll_fd,wait_event,SOMB_WORK_POLLWAIT_ITEM_MAX,-1);
        if(ret < 0)
        {
            SOMB_LOG_ERR("somb work-proxy:(%s) SubNsId(%d) pid(%d) epoll_wait ret=%d errno=%s",pstLocalSubNs->localexename, pstLocalSubNs->sub_ns_id,pstLocalSubNs->localpid,ret,strerror(errno));
            continue;
        }
        else if(ret == 0) /* timeout */
        {
            SOMB_LOG_ERR("somb work-proxy:(%s) SubNsId(%d) pid(%d) epoll_wait ret=%d timeout! errno=%s",pstLocalSubNs->localexename,
                pstLocalSubNs->sub_ns_id,pstLocalSubNs->localpid,ret,strerror(errno));
            continue;
        }

        SOMB_LOG_DEBUG("somb work-proxy:(%s) SubNsId(%d) pid(%d) epoll_wait ret=%d ",pstLocalSubNs->localexename,
                pstLocalSubNs->sub_ns_id,pstLocalSubNs->localpid,ret);
        uint8_t loop = 0;
        for(;loop < ret;loop++)
        {
            if((wait_event[loop].events & EPOLLIN) == EPOLLIN)
            {
                if(wait_event[loop].data.fd == pstLocalSubNs->sk1)
                {
                    //远端的client连接
                    vsomb_worker_proxy_sk1_process(epoll_fd,pstLocalSubNs);
                }
                else if((wait_event[loop].data.fd == pstLocalSubNs->sk)||(wait_event[loop].data.fd == pstLocalSubNs->ctrl_sk))
                {
                    //从somb_daemon or somb_ctrl 接收消息处理
                    vsomb_worker_proxy_daemon_or_ctrl_process(wait_event[loop].data.fd,epoll_fd,pstLocalSubNs);
                }
                else
                {
                    //处理远端的client发送来消息
                    vsomb_worker_proxy_business_process(epoll_fd,pstLocalSubNs,wait_event[loop].data.fd);
                }
            }
            else if((wait_event[loop].events & EPOLLRDHUP) == EPOLLRDHUP)
            {
                struct somb_mngr_sub_ns *pstSubNs = (struct somb_mngr_sub_ns *)(wait_event[loop].data.ptr);
                pstSubNs->SubNsStatus = SOMB_SUBNS_STATUS_OFFLINE;
                ret = epoll_ctl(epoll_fd, EPOLL_CTL_DEL, wait_event[loop].data.fd, NULL);
                SOMB_LOG_WARN("somb work-proxy:(%s) SubNsId(%d) pid(%d) del client fd(%d) of (%s-%d) ",pstLocalSubNs->localexename, pstLocalSubNs->sub_ns_id,pstLocalSubNs->localpid,wait_event[loop].data.fd ,pstSubNs->sub_ns_name,pstSubNs->sub_ns_id);
            }
            else
            {
                SOMB_LOG_ERR("somb work-proxy:(%s) SubNsId(%d) pid(%d) recv fd=%d ret=%d",pstLocalSubNs->localexename, pstLocalSubNs->sub_ns_id,pstLocalSubNs->localpid, wait_event[loop].data.fd,ret);
            }
        }
    }
}

int32_t SOMB_Worker_Init(void)
{
    int ret;

    pthread_attr_t attr;
    pthread_attr_init(&attr);

    pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);

    ret = pthread_create(&(gstSombWork[0].tid), &attr,vsomb_worker_proxy ,NULL);
    if(0 != ret)
    {
        SOMB_LOG_ERR("somb work:pid(%d) create proxy thread failled",getpid());
        ret = SOMB_FAIL;
    }


    pthread_attr_destroy(&attr);


    return ret;
}
